package cn.com.sparkle.paxos.net;

import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.deamon.ReConnectThread.ReConnectMethod;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.NodeStateChangeEvent;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SenatorHeatBeatRequest;
import cn.com.sparkle.paxos.net.callback.HeatBeatCallBack;
import cn.com.sparkle.paxos.net.factory.NetClient;
import cn.com.sparkle.paxos.net.factory.NetHandler;
import cn.com.sparkle.paxos.net.factory.PaxosSession;

public class ClientNetHandler implements NetHandler {
	private final static Logger logger = Logger
			.getLogger(ClientNetHandler.class);

//	private final NioSocketClient client;
//	private IoHandler handler;
	private final NetClient netClient;
	private ReentrantLock lock = new ReentrantLock();
	private EventsManager eventsManager;
	private ReConnectThread reConnectThread;

	public ClientNetHandler(NetClient netClient,
			EventsManager eventsManager, ReConnectThread reConnectThread) {
		this.netClient = netClient;
		this.eventsManager = eventsManager;
		this.reConnectThread = reConnectThread;
	}
	
	@Override
	public void onDisconnect(PaxosSession session) {
		NetNode nnode = (NetNode) session.attachment();
		nnode.nodeState.setConnected(false);
		try {
			lock.lock();
			NodeStateChangeEvent.doLoseConnectEvent(eventsManager, nnode);
		} finally {
			lock.unlock();
		}
		nnode.close();
		reConnect(nnode.nodeState);
	}


	@Override
	public void onConnect(PaxosSession session) {
		logger.debug("open connected session" + session.toString());
		NodeState nodeState = (NodeState) session.attachment();
		nodeState.setConnected(true);
		NetNode nnode = new NetNode(session, nodeState);
		
		SenatorHeatBeatRequest request = PaxosMessages.SenatorHeatBeatRequest
				.getDefaultInstance();
		nnode.write(request, new HeatBeatCallBack(eventsManager){
			@Override
			public void call(NetNode nnode, MessagePackage value) {
				super.call(nnode, value);
				try {
					lock.lock();
					NodeStateChangeEvent.doOpenConnectEvent(eventsManager, nnode);
				} finally {
					lock.unlock();
				}
			}
		});
		session.attach(nnode);
	}


	@Override
	public void onRecieve(PaxosSession session, MessagePackage message) {
		NetNode nnode = (NetNode) session.attachment();
		nnode.recieve(message);
	}

	@Override
	public void onRefuse(Object connectAttachment) {
		NodeState ns = (NodeState)connectAttachment;
		reConnect(ns);
	}


	private final class ReConnect implements ReConnectMethod {
		private final Logger log = Logger.getLogger(ReConnect.class);

		@Override
		public void reConnect(Object value) {
			NodeState ns = (NodeState) value;
			String address = ns.getAddress();
			String[] a = address.split(":");
			try {
				log.debug("reConnect " + address);
				netClient.connect(a[0], Integer.parseInt(a[1]), ns);
			} catch (Throwable e) {
				reConnectThread.add(ns, this, 2000);
			}
		}

	}

	private final ReConnect method = new ReConnect();

	private void reConnect(NodeState ns) {
		// After sleep 5 seconds,retry to connect.
		reConnectThread.add(ns, method, 5000);
	}


	

}
